package com.boot.comsumer;

import com.alibaba.fastjson.JSONObject;
import com.boot.entity.Config;
import com.boot.entity.config_canal.ConfigCanalBean;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.rest.RestStatus;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * kafka消费者（监听名为canal-test-topic的topic），同步ElasticSearch
 * @author youzhengjie 2022-09-01 16:54:28
 */
@Component
@Slf4j
public class ConfigCanalElasticSearchConsumer {

    @Autowired
    private RestHighLevelClient restHighLevelClient;

    //es的index，相当于mysql的数据库：（数据库.表名）
    private static final String ES_INDEX = "config-canal-es";

    //过期时间（单位：小时）
    private static final int TIME_OUT = 24;


    /**
     * @param consumer 接收消费记录（消息）
     * @param ack 手动提交消息
     */
    @KafkaListener(topics = "canal-test-topic")
    public void receive(ConsumerRecord<String, String> consumer, Acknowledgment ack) {

        try {
            //获取canal的消息
            String value = (String) consumer.value();
            log.info("topic名称:{},key:{},分区位置:{},下标:{},value:{}", consumer.topic(), consumer.key(), consumer.partition(), consumer.offset(), value);

            //转换为javaBean
            ConfigCanalBean canalBean = JSONObject.parseObject(value, ConfigCanalBean.class);
            /*
            由于我们canal.instance配置了监控canal-test-db1.t_config表和canal-test-db1.t_user表（生产环境下可以启动多个canal，每一个canal监听一张需要同步的表）
            所以我们要对这两张表分开处理。（可以通过他们的表名（canalBean.getTable()）来区分）
            如果canalBean.getTable()获取的表名是t_config，则同步到es，如果不是则不管。
             */
            log.warn("["+canalBean+"]");
            if("t_config".equals(canalBean.getTable())){
                //获取是否是DDL语句
                boolean isDdl = canalBean.isDdl();
                //获取当前sql语句的类型（比如INSERT、DELETE等等）
                String type = canalBean.getType();
                List<Config> datas = canalBean.getData();
                if ("t_config".equals(canalBean.getTable())) {
                    //如果不是DDL语句
                    if (!isDdl) {
                        //INSERT和UPDATE都是一样的操作
                        if ("INSERT".equals(type) || "UPDATE".equals(type)) {
                            //新增语句
                            for (Config config : datas) {
                                // 增加、修改处理
                                IndexRequest indexRequest = new IndexRequest(ES_INDEX);
                                indexRequest.id(config.getConfigId()+""); //id
                                ConcurrentHashMap<String, Object> dataMap = new ConcurrentHashMap<>();
                                dataMap.put("configInfo",(config.getConfigInfo()!=null)?config.getConfigInfo():"");
                                dataMap.put("datetime",(config.getDatetime()!=null)?config.getDatetime():"");
                                dataMap.put("desc",(config.getDesc()!=null)?config.getDesc():"");
                                indexRequest.source(dataMap);

                                IndexResponse response= restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);

                                RestStatus status = response.status();
                                log.info("status="+status.toString());
                            }
                        }else if("DELETE".equals(type)){
                            //删除语句
                            if(datas!=null && datas.size()>0){
                                for (Config config : datas) {
                                    DeleteRequest deleteRequest = new DeleteRequest();
                                    deleteRequest.id(config.getConfigId()+"");
                                    DeleteResponse deleteResponse = restHighLevelClient.delete(deleteRequest, RequestOptions.DEFAULT);
                                    log.info("deleteResponse:"+deleteResponse);
                                }
                            }
                        }

                    }
                }
            }
            //最后，如果上面的代码没有报错的情况下，可以确认消息了。（很重要）
            ack.acknowledge();
        }catch (Exception e){
            throw new RuntimeException();
        }
    }

}

